package org.yuanzheng.tableAndSql

import org.apache.flink.streaming.api.TimeCharacteristic
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.table.api.scala.StreamTableEnvironment
import org.apache.flink.table.api.{EnvironmentSettings, Table, Tumble}
import org.apache.flink.types.Row
import org.yuanzheng.source.StationLog

/*tumble窗口*/
/*每隔5秒钟统计，每个基站的通话数量，假设数据是乱序，最多延时3秒（watermark）*/
object TestWindowBySql {
  def main(args: Array[String]): Unit = {
    //创建使用flink原生
    val streamEnv: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    streamEnv.setStreamTimeCharacteristic(TimeCharacteristic.EventTime)
    val settings: EnvironmentSettings = EnvironmentSettings.newInstance().useOldPlanner().inStreamingMode().build()
    val tableEnv: StreamTableEnvironment = StreamTableEnvironment.create(streamEnv, settings)

    //隐式转换
    import org.apache.flink.streaming.api.scala._
    import org.apache.flink.table.api.scala._

    //读取数据
    val stream: DataStream[StationLog] = streamEnv.socketTextStream("192.168.1.10", 8888)
      .map(line => {
        var split = line.split(",")
        new StationLog(split(0).trim, split(1).trim, split(2).trim, split(3).trim, split(4).trim.toLong, split(5).trim.toLong)
      })
      //引入水位线，延迟触发
      .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[StationLog](Time.seconds(3)) {
        override def extractTimestamp(element: StationLog): Long = {
          element.callTime
        }
      })

    //创建动态table并指定EventTime字段(使用.rowtime)
    tableEnv.registerDataStream("t_station_log", stream, 'sid, 'callOut, 'callIn, 'callType, 'callTime.rowtime, 'duration)

    //使用SQL进行开窗聚合
    val result: Table = tableEnv.sqlQuery("select sid, sum(duration) as sd from t_station_log where callType='success' group by tumble(callTime,interval '5' second), sid")

    tableEnv.toRetractStream[Row](result).filter(_._1 == true).print()
    streamEnv.execute()
  }
}
